/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.ws.cli;

import org.apache.commons.cli.*;

/**
 * @author wangshuai
 */
public class CliOptionsParser {
    public static final String MY_CATALOG = "myCatalog";
    public static final String DEFAULT = "default";

    public static final Option OPTION_WORKING_SPACE = Option
            .builder("w")
            .required(false)
            .longOpt("working_space")
            .numberOfArgs(1)
            .argName("working space dir")
            .desc("The working space dir.the default System.getenv(\"FLINK_HOME\")")
            .build();

    public static final Option OPTION_SQL_FILE = Option
            .builder("r")
            .required(false)
            .longOpt("regular")
            .numberOfArgs(1)
            .argName("SQL file name")
            .desc("The SQL file path,support regular matching. the default flink.sql")
            .build();
    public static final Option OPTION_HIVE_CATALOG = Option
            .builder("c")
            .required(false)
            .longOpt("hive_config")
            .numberOfArgs(1)
            .argName("hive config")
            .desc("hive configuration folder address")
            .build();
    public static final Option OPTION_DEFAULT_CATALOG = Option
            .builder("l")
            .required(false)
            .longOpt("catalog")
            .numberOfArgs(1)
            .argName("catalog")
            .desc("specify the current catalog, the default catalog is my catalog")
            .build();
    public static final Option OPTION_DEFAULT_DATABASE = Option
            .builder("d")
            .required(false)
            .longOpt("database")
            .numberOfArgs(1)
            .argName("database")
            .desc("Specify the current data to be used, the default is default")
            .build();
    public static final Option OPTION_HELP =
            Option.builder("h")
                    .required(false)
                    .longOpt("help")
                    .desc("Show the help message with descriptions of all options.")
                    .build();
    public static final Options CLIENT_OPTIONS = getClientOptions(new Options());

    public static Options getClientOptions(Options options) {
        options.addOption(OPTION_SQL_FILE);
        options.addOption(OPTION_WORKING_SPACE);
        options.addOption(OPTION_HIVE_CATALOG);
        options.addOption(OPTION_DEFAULT_CATALOG);
        options.addOption(OPTION_DEFAULT_DATABASE);
        options.addOption(OPTION_HELP);
        return options;
    }

    // --------------------------------------------------------------------------------------------
    //  Line Parsing
    // --------------------------------------------------------------------------------------------

    public static CliOptions parseClient(String[] args) {
        try {
            DefaultParser parser = new DefaultParser();
            CommandLine line = parser.parse(CLIENT_OPTIONS, args, true);
            if (line.hasOption(CliOptionsParser.OPTION_HELP.getOpt())) {
                HelpFormatter formatter = new HelpFormatter();
                formatter.setLeftPadding(5);
                formatter.setWidth(80);

                System.out.println(
                        "flink sql file submission parameter description");
                formatter.setSyntaxPrefix("  \"submit\" mode options:");
                formatter.printHelp(" ", CLIENT_OPTIONS);
                System.exit(0);
            }
            String sqlFile = line.getOptionValue(CliOptionsParser.OPTION_SQL_FILE.getOpt(), "flink.sql");
            String workPath = line.getOptionValue(CliOptionsParser.OPTION_WORKING_SPACE.getOpt(), System.getenv("FLINK_HOME"));
            String defaultCatalog = line.getOptionValue(CliOptionsParser.OPTION_DEFAULT_CATALOG.getOpt(), MY_CATALOG);
            String database = line.getOptionValue(CliOptionsParser.OPTION_DEFAULT_DATABASE.getOpt(), DEFAULT);
            return new CliOptions(
                    sqlFile,
                    workPath,
                    line.getOptionValue(CliOptionsParser.OPTION_HIVE_CATALOG.getOpt()),
                    defaultCatalog,
                    database
            );
        } catch (ParseException e) {
            throw new RuntimeException(e.getMessage());
        }
    }
}
